package com.frinder.hadoop.job.joiner;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * @author frinder
 * @date 2018/6/26
 * @Description: ${todo}
 */
@Slf4j
public class TextPair implements WritableComparable<TextPair> {

    private Text first;
    private Text second;

    public TextPair() {
        set(new Text(), new Text());
    }

    public TextPair(String first, String second) {
        set(new Text(first), new Text(second));
    }

    public void set(Text first, Text second) {
        this.first = first;
        this.second = second;
    }

    public Text getFirst() {
        return first;
    }

    public Text getSecond() {
        return second;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        first.write(out);
        second.write(out);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        first.readFields(in);
        second.readFields(in);
    }

    @Override
    public int compareTo(TextPair tp) {
        int result = first.compareTo(tp.first);
        if (result == 0) {
            result = second.compareTo(tp.second);
        }
        log.info("*** compareTo 1 result: {}", result);
        return result;
    }

    @Override
    public boolean equals(Object o) {
        if (o instanceof TextPair) {
            TextPair tp = (TextPair) o;
            return first.equals(tp.first) && second.equals(tp.second);
        }
        return false;
    }

    @Override
    public int hashCode() {
        int result = first.hashCode() * 163 + second.hashCode();
        return result;
    }

    @Override
    public String toString() {
        return "TextPair{" +
                "first=" + first +
                ", second=" + second +
                '}';
    }

    /**
     * byte比较器，效率高于 compareTo，优先于 compareTo执行
     */
    public static class Comparator extends WritableComparator {

        private static final Text.Comparator TEXT_COMPARATOR = new Text.Comparator();

        public Comparator() {
            super(TextPair.class);
        }

        @Override
        public int compare(byte[] b1, int s1, int l1,
                           byte[] b2, int s2, int l2) {
            try {
                int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
                int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
                int result = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
                if (result == 0) {
                    result = TEXT_COMPARATOR.compare(b1, s1 + firstL1, l1 - firstL1,
                            b2, s2 + firstL2, l2 - firstL2);
                    log.info("*** compare->first is equal, result:{}", result);
                } else {
                    log.info("*** compare->first is not equal, result:{}", result);
                }
                return result;
            } catch (IOException e) {
                throw new IllegalArgumentException(e);
            }
        }
    }

    static {
        WritableComparator.define(TextPair.class, new Comparator());
    }

    /**
     * 分组比较器，供map后根据key分组使用
     */
    public static class FirstComparator extends WritableComparator {

        private static final Text.Comparator TEXT_COMPARATOR = new Text.Comparator();

        public FirstComparator() {
            super(TextPair.class);
        }

        @Override
        public int compare(byte[] b1, int s1, int l1,
                           byte[] b2, int s2, int l2) {
            try {
                int firstL1 = WritableUtils.decodeVIntSize(b1[s1]) + readVInt(b1, s1);
                int firstL2 = WritableUtils.decodeVIntSize(b2[s2]) + readVInt(b2, s2);
                int result = TEXT_COMPARATOR.compare(b1, s1, firstL1, b2, s2, firstL2);
                log.info("*** FirstComparator result 1: {}", result);
                return result;
            } catch (IOException e) {
                throw new IllegalArgumentException(e);
            }
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            if (a instanceof TextPair && b instanceof TextPair) {
                int result = ((TextPair) a).first.compareTo(((TextPair) b).first);
                log.info("*** FirstComparator result 2: {}", result);
                return result;
            }
            return super.compare(a, b);
        }
    }
}
